package f8watermark;

import orderdata.OrderData;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;

public class Wm {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> scostr = env.socketTextStream("192.168.43.102", 7777);
        //端口读取数据内容映射为订单类
        scostr.map(new MapFunction<String, OrderData>() {
            @Override
            public OrderData map(String s) throws Exception {
                String[] arr = s.split(",");
                return new OrderData(
                        Long.parseLong(arr[0]),   // order_date (timestamp)
                        arr[1],                   // category
                        arr[2],                   // region
                        Integer.parseInt(arr[3]), // user_id
                        Integer.parseInt(arr[4]), // order_count
                        Double.parseDouble(arr[5])// total_amount
                );
            }
        })
                .assignTimestampsAndWatermarks(//设置事件时间时间戳和 watermark
                        WatermarkStrategy.
                                <OrderData>forBoundedOutOfOrderness(Duration.ofSeconds(5))//最多允许数据迟到 5 秒
                                .withTimestampAssigner(//时间戳分配器
                                        new SerializableTimestampAssigner<OrderData>() {//将事件数据流中的每个事件分配一个时间戳

                                            @Override
                                            public long extractTimestamp(OrderData orderData, long l) {
                                                return orderData.order_date;
                                            }
                                        })
                );

    }
}
